//
// Created by Administrator on 2021/4/7.
//
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include <amqp.h>
#include <amqp_tcp_socket.h>

#include "rabbitmq_api.h"
#include "rabbitmq_utils.h"
#include "config.h"
#include "head.h"

// extern struct Config config;

amqp_connection_state_t connectServer(char *ip, int port) {
    int status;
    amqp_connection_state_t conn = NULL;
    amqp_socket_t *socket = NULL;
    conn = amqp_new_connection();

    socket = amqp_tcp_socket_new(conn);
    if (!socket) {
        return NULL;
    }

    status = amqp_socket_open(socket, ip, port);
    if (status) {
        return NULL;
    }

    amqp_rpc_reply_t loginRes = amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "admin", "admin");
    if (loginRes.reply_type != AMQP_RESPONSE_NORMAL) {
        die_on_amqp_error(loginRes, "Logging in");
        return NULL;
    }
    // amqp_channel_open(conn, channel);
    // amqp_rpc_reply_t replyRes = amqp_get_rpc_reply(conn);
    // if (replyRes.reply_type != AMQP_RESPONSE_NORMAL) {
    //     return NULL;
    // }

    return conn;
}

// amqp_connection_state_t getPublishConn(char *ip, int port, char *queueName) {
//     amqp_connection_state_t conn = connectServer(ip, port);
//     if (conn == NULL) {
//         return NULL;
//     }
//     amqp_channel_open(conn, 1);
//     amqp_rpc_reply_t replyRes = amqp_get_rpc_reply(conn);
//     if (replyRes.reply_type != AMQP_RESPONSE_NORMAL) {
//         return NULL;
//     }
//     amqp_basic_consume(conn, 1, amqp_cstring_bytes(queueName), amqp_empty_bytes,
//                        0, 1, 0, amqp_empty_table);
//     die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");
// }

amqp_connection_state_t getRecvConn(char *ip, int port, char **queueName) {
    amqp_connection_state_t conn = connectServer(ip, port);
    if (conn == NULL) {
        return NULL;
    }
    amqp_channel_open(conn, 1);
    amqp_rpc_reply_t replyRes = amqp_get_rpc_reply(conn);
    if (replyRes.reply_type != AMQP_RESPONSE_NORMAL) {
        return NULL;
    }
    for (int i = 2; i < 3; i++) {
        amqp_basic_consume(conn, 1, amqp_cstring_bytes(queueName[i]), amqp_empty_bytes,
                           0, 0, 0, amqp_empty_table);
        die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");
    }

}

int publishMessage(amqp_connection_state_t conn, int channel, int type, char *hostName, char *ip, char *message) {
    if (conn == NULL) {
        return 1;
        // int ret = 0;
        // if ((ret = connectServer()) != 0) {
        //    DBG("<"RED"ERROR"NONE"> : connect rabbit server failed, ret=%d", ret);
        // }
    }
    char routingKey[128] = {0};
    switch (type) {
        case 1:
            sprintf(routingKey, "cpu.%s.%s", hostName, ip);
            break;
        case 2:
            sprintf(routingKey, "mem.%s.%s", hostName, ip);
            break;
        case 3:
            sprintf(routingKey, "disk.%s.%s", hostName, ip);
            break;
        case 4:
            sprintf(routingKey, "proc.%s.%s", hostName, ip);
            break;
        case 5:
            sprintf(routingKey, "sys.%s.%s", hostName, ip);
            break;
        case 6:
            sprintf(routingKey, "user.%s.%s", hostName, ip);
            break;
        default:
            break;
    }
    amqp_basic_properties_t props;
    props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG;
    props.content_type = amqp_cstring_bytes("text/plain");
    props.delivery_mode = 2; /* persistent delivery mode */

    DBG("<"GREEN"INFO"NONE"> : send messge [%s], routingKey: %s", message, routingKey);
    int publishRes = AMQP_STATUS_OK;
    if (amqp_basic_publish(conn, channel, amqp_cstring_bytes(EXCHANGENAME), amqp_cstring_bytes(routingKey), 0, 0, &props, amqp_cstring_bytes(message)) != AMQP_STATUS_OK) {
        // DBG("<"RED"ERROE"NONE"> : send messge [%s] failed", message);
        die_on_error(publishRes, "PublishMessage");
        return 2;
    }
//    die_on_error(amqp_basic_publish(conn, 1, amqp_cstring_bytes(EXCHANGENAME), amqp_cstring_bytes(routingKey), 0, 0, &props, amqp_cstring_bytes(message)),
//                 "Publishing");
    return 0;
}

void amqp_get_hostname(amqp_envelope_t *envelope, char *hostname) {
    char *routingKey = calloc(envelope->routing_key.len + 1, sizeof(char));
    strncpy(routingKey, envelope->routing_key.bytes, envelope->routing_key.len);
    char *indStart = strchr(routingKey, '.');
    char *indEnd = strchr(indStart + sizeof(char), '.');
    strncpy(hostname, indStart + sizeof(char), indEnd - indStart - 1);
    free(routingKey);
}

void amqp_get_ip(amqp_envelope_t *envelope, char *ip) {
    char *routingKey = calloc(envelope->routing_key.len + 1, sizeof(char));
    strncpy(routingKey, envelope->routing_key.bytes, envelope->routing_key.len);
    // char *routingKey = envelope->routing_key.bytes;
    char *tmpIp;
    tmpIp = strchr(routingKey, '.');
    tmpIp += sizeof(char);
    tmpIp = strchr(tmpIp, '.');
    strcpy(ip, tmpIp + sizeof(char));
    free(routingKey);
}

int recvMessage(amqp_connection_state_t conn, int channel, char *queueName, char *message, int length, char *hostname, char *ip) {
    if (conn == NULL) {
        return 1;
        // int ret = 0;
        // if ((ret = connectServer()) != 0) {
        //     DBG("<"RED"ERROR"NONE"> : connect rabbit server failed, ret=%d", ret);
        // }
    }

    amqp_rpc_reply_t res;
    amqp_envelope_t envelope;

    amqp_basic_consume(conn, 1, amqp_cstring_bytes(queueName), amqp_empty_bytes,
                       0, 0, 0, amqp_empty_table);
    die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");

    amqp_maybe_release_buffers(conn);
    // Recv Interval
    struct timeval timeInter;
    timeInter.tv_sec = 0;
    timeInter.tv_usec = 0;
    // nonblock recv message
    res = amqp_consume_message(conn, &envelope, NULL, 0);
    if (AMQP_RESPONSE_NORMAL != res.reply_type) {
        return 1;
    }
    if (envelope.message.body.len > length) {
        return 2;
    }
    amqp_get_hostname(&envelope, hostname);
    amqp_get_ip(&envelope, ip);
    strcpy(message, envelope.message.body.bytes);
    // 手动应答
    amqp_basic_ack(conn, 1, envelope.delivery_tag, 0);
    amqp_destroy_envelope(&envelope);

    return 0;
}

